import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PrintWriter;
import java.io.Serializable;
import java.io.StreamCorruptedException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.HashMap;


/**
 * This class is designed to organize and coordinate all the different
 * nodes on the Chord ring.  Each node will have its own finger table.  
 * 
 * @author Guillermo Mendez-Kestler
 * @author Patrick Ramsey
 * @version 0.1
 */
public class Node {
	HashMap<String, Serializable> locObj;
	Socket socketIn;
	Socket socketOut;
	ServerSocket serverSockIn;
	ServerSocket serverSockOut;
	Socket socket;
	ServerSocket serverSock;
	public static final int DEFAULT_PORT = 2793;
	public static final int DEFAULT_PORT_IN = 9685;
	public static final int DEFAULT_PORT_OUT = 9437;
	private static final boolean DEBUG = true;
	private ObjectOutputStream out;
	private ObjectInputStream in;
	private FingerTable ft;
	private Object lookupReturn;
	
	/**
	 * Creates a node to be placed onto the ring. 
	 * 
	 * @param host The hostname of the node
	 * @param port The port number of the node
	 * @throws IOException
	 */
	public Node(InetAddress host, int port) throws IOException {
		locObj = new HashMap<String, Serializable>();
		serverSock = new ServerSocket(DEFAULT_PORT);
		serverSockIn = new ServerSocket(DEFAULT_PORT_IN);
		serverSockOut = new ServerSocket(DEFAULT_PORT_OUT);
		ft = new FingerTable(host.getHostName()+":"+port);
		
	}//Node
	
	/**
	 * Creates and joins the newly created ring
	 */
	public void joinRing() {
		FingerTable.arrays(ft);
		while(true)
		{
			try{
				socket = serverSock.accept();
				//socketIn = serverSockIn.accept();
				//socketOut = serverSockOut.accept();
			}
			catch(Exception e)
			{
				System.err.println(e);
			}
			Runnable c = new myRunnable();
			Thread t1 = new Thread(c);
			t1.start();
		}//while
	}
	
	/**
	 * Joins the ring with the IP address and port of a node already 
	 * on the ring. 
	 * 
	 * @param ip IP Address of the node already on the ring
	 * @param port Port of the node already on the ring
	 */
	@SuppressWarnings("unchecked")
	public void joinRing(String ip, int port) {
		debug("In join already created ring");
		PrintWriter outPW;
		Socket sock;
		
		try{
			debug("before creating socket");
			sock = new Socket(ip, port);
			debug("before creating ObjectOutputStream");
			out = new ObjectOutputStream(sock.getOutputStream());
			debug("in = new ObjectInputStream(sock.getInputStream());");
			//in = new ObjectInputStream(sock.getInputStream());
			debug("before creating outputwriter");
			outPW = new PrintWriter(sock.getOutputStream(), true);
			//this gets the finger table of the node we specify to get its finger table to help establish our finger table
			debug("before -getFT");
			outPW.println("-getFT " + InetAddress.getLocalHost()+":"+Node.DEFAULT_PORT );
			debug("after -getFT");
			outPW.println("-printFT");
			//unsafe cast but has to be done
			debug("before copyFinger");
			outPW.println("-copyFT");
			in = new ObjectInputStream(sock.getInputStream());
			ft.copyFinger((HashMap<Integer, String>) in.readObject());
			//this is sent around the network to say that we are a new node
			debug("attempting to send addFT around the network");
			outPW.println("-addFT " + InetAddress.getLocalHost()+":"+Node.DEFAULT_PORT);
			out.close();
			sock.close();
			//this is called to make sure we fix our own finger table
			this.fixFinger();
			
		} catch (StreamCorruptedException	e){
			System.err.println(e);
			e.printStackTrace();
		} catch (IOException e){
			System.err.println(e);
			e.printStackTrace();
		} catch (SecurityException e){
			System.err.println(e);
			e.printStackTrace();
		} catch (NullPointerException e){
			System.err.println(e);
			e.printStackTrace();
		} catch (ClassNotFoundException e) {
			System.err.println(e);
			e.printStackTrace();
		}
		
		while(true)
		{
			try{
				socket = serverSock.accept();
				//socketIn = serverSockIn.accept();
				//socketOut = serverSockOut.accept();
			}
			catch(Exception e)
			{
				System.err.println(e);
			}
			Runnable c = new myRunnable();
			Thread t1 = new Thread(c);
			t1.start();
		}//while
		
		
	}
	
	/**
	 * Leaves the ring and updates the two surrounding nodes. 
	 * To be implemented in the future
	 */
	public void leaveRing() {
		
	}
	
	/**
	 * Finds the next node in the ring. 
	 * @return The next node in the ring. 
	 */
	public String[] findSuccessor() {
		return ft.get(1).split(":");
	}
	
	/**
	 * Finds the previous node in the ring.  
	 * @return The previous node in the ring.  
	 */
	public String[] findPredecessor() {
		return ft.get(0).split(":");
	}
	
	/**
	 * Adds an Object to the Ring
	 * @param key The String which describes value
	 * @param value The Object itself
	 */
	public void addObject(String key,Serializable value) {
		int hashKey = this.hashCode(key);
		
		//if we are responsible for it put it in our own table
		if(hashKey < this.hashCode(findSuccessor().toString()))
		{
			locObj.put(key, value);
		}
		
		String[] hostNamePort = ft.get(hashKey).split(":");
		try{
			socket = new Socket(hostNamePort[0], Integer.parseInt(hostNamePort[1]));
			out = new ObjectOutputStream(new BufferedOutputStream(socket.getOutputStream()));
			out.writeObject("-put");
			out.writeObject(key);
			out.writeObject(value);
		}
		catch(Exception e)
		{
			System.err.println(e);
		}
		
		try {
			out.close();
			socket.close();
		} catch (IOException e1) {
			e1.printStackTrace();
		}
		
	}
	
	/**
	 * Updates all nodes for when nodes join or leave the ring
	 * TO be implemented in the future
	 */
	public void stabalizeRing() {
		
	}
	
	/**
	 * Updates the Finger Table for the current node. 
	 */
	public void fixFinger() {
		boolean suc = true, pre = true;
		// Obtain the successor
		String[] succ = findSuccessor();
		try {
			// OPEN CONNECTIONS TO SUCCESSOR
			socket = new Socket(succ[0], Integer.parseInt(succ[1]));
			in = new ObjectInputStream(socket.getInputStream());
			out = new ObjectOutputStream(socket.getOutputStream());
			// SEE IF WE CAN COMMUNICATE BETWEEN EACH OTHER
			chitChat(in, out);/****************************************************/
			// CLOSE CONNECTIONS TO SUCCESSOR
			out.close();
			in.close();
			socket.close();
		} catch (NumberFormatException e) {
			e.printStackTrace();
		} catch (UnknownHostException e) {
			e.printStackTrace();
			suc = false;
			System.out.println("\"" + succ[0] + ":" + succ[1] + "\" was not found. ");
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		// Obtain the predecessor
		String[] pred = findPredecessor();
		try {
			// OPEN CONNECTIONS TO PREDECESSOR
			socket = new Socket(pred[0], Integer.parseInt(pred[1]));
			in = new ObjectInputStream(socket.getInputStream());
			out = new ObjectOutputStream(socket.getOutputStream());
			// SEE IF WE CAN COMMUNICATE BETWEEN EACH OTHER
			chitChat(in, out);/****************************************************/
			// CLOSE CONNECTIONS TO PREDECESSOR
			out.close();
			in.close();
			socket.close();
		} catch (NumberFormatException e) {
			e.printStackTrace();
		} catch (UnknownHostException e) {
			e.printStackTrace();
			pre = false;
			System.out.println("\"" + pred[0] + ":" + pred[1] + "\" was not found. ");
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		// Checks to see if there was communication
		if (!suc){
			ft.remove(succ[0] + ":" + succ[1]);
		}
		if (!pre){
			ft.remove(pred[0] + ":" + pred[1]);
		}
	} // fixFinger()
	
	private void chitChat(ObjectInputStream ccIn, ObjectOutputStream ccOut) 
			throws IOException {
		ccOut.writeInt(1);
		ccIn.read();
	} // chitChat(OIS, OOS)
	
	/**
	 * Lookup the Object you want to find
	 * 
	 * @param key the object you want to find
	 * @return the object you wanted in the first place, or null
	 *           if the object is not in the ring. 
	 */
	public boolean lookup(String key, String hostNamePort) {
		int hashKey = this.hashCode(key);
		Object val = null;
		if(hashKey < this.hashCode(findSuccessor().toString()))
		{
			val = locObj.get(key);
		}
		//if we found something send it back to the origin.
		if (val != null)
		{
			String[] str = hostNamePort.split(":");
			try{
				socket = new Socket(str[0],Integer.parseInt(str[1]));
				out = new ObjectOutputStream(new BufferedOutputStream(socket.getOutputStream()));
				out.writeObject("-reply");
				out.writeObject(val);
				return true;
				
			}
			catch(Exception e)
			{
				System.err.println(e);
			}
		}
		else
		{
			String[] str = ft.get(hashKey).split(":");
			try {
				socket = new Socket(str[0],Integer.parseInt(str[1]));
				out = new ObjectOutputStream(new BufferedOutputStream(socket.getOutputStream()));
				out.writeObject("-lookup " + key + " " + hostNamePort);
			} catch (Exception e) {
				System.err.println(e);
			}
		}
		
		try{
			out.close();
			socket.close();
		}
		catch(Exception e)
		{
			System.err.println(e);
		}
		
		return false;

	}
	/**
	 * this method gets the variable lookupReturn that is returned from a lookup call
	 * @return
	 */
	public Object GetLookupReturn()
	{
		return lookupReturn;
	}
	
	/**
	 * Obtain the hash code for the local "hostname:port" of this node. 
	 * @return the hash code for the local "hostname:port"
	 */
	public int hashCode(Object o) {
		int x = o.hashCode();
		x = x & 0xffff;
		return x;
	}
	
	/**
	 * Prints out debugging information when DEBUG is set to true. 
	 * @param msg The message to be printed if DEBUG is true. 
	 */
	private void debug(String msg) {
		if(DEBUG) System.out.println(msg);
	}
	/**
	 * this subclass deals with the message passing
	 * around the ring and is threaded. Messages that can be
	 * Received are 
	 * -l for lookup
	 * -f for fix finger table. Called when a new Node joins
	 * -p for putting objects in the table. The next two messages
	 * after -p are the object key and the object value
	 * @author Pat
	 *
	 */
	class myRunnable implements Runnable
	{
		int objectRecievedCount;
		Object key;
		Object value;
		int seenAddFT;
		boolean addFT;
		public myRunnable()
		{
			objectRecievedCount = 0;
			key = null;
			value = null;
			seenAddFT = 0;
			addFT = true;
		}
		
		
		
		public void run()
		{
			try
			{
				
				out = new ObjectOutputStream(new BufferedOutputStream(socket.getOutputStream()));
				in = new ObjectInputStream(new BufferedInputStream(socket.getInputStream()));
				
				String input = (String)in.readObject();
				debug("getting the input and converting it to a string");
				if( (input == null))
				{
					//kick out of loop
					debug("in null input");
					return;
				}
				synchronized(this)
				{
					String[] str = input.split(" ");
					
					if(str[0].equalsIgnoreCase("-lookup"))
					{
						debug("lookup has been crecieved");
						lookup(str[1], str[2]);
					}
					if(str[0].equalsIgnoreCase("-reply"))
					{
						debug("reply has been recieved");
						this.nextMessageIsObject();
					}
						
					if(str[0].equalsIgnoreCase("-put"))
					{
						debug("put has been recieved");
						this.nextTwoMessagesAreObjects();
					}
						
					if(str[0].equalsIgnoreCase("-addFT"))
					{
						debug("addFT has been recieved");
						ft.add(str[1]);
						this.sendAddFT(input);
					}
					
					if(str[0].equalsIgnoreCase("-getFT"))
					{
						debug("getFT has been recieved");
						sendFingerTable(str[1]);
					}
					
					if(str[0].equalsIgnoreCase("-copyFT"))
					{
						debug("copyFT has been recieved");
						out.writeObject(ft.getFinger());
					}
					
					if(str[0].equalsIgnoreCase("-printFT"))
					{
						debug("printing finger table");
						FingerTable.arrays(ft);
					}
						
				}//sync
				
				debug("closing input and output streams and socket");
				in.close();
				out.close();
				socket.close();
				
			}
			catch(Exception e)
			{
				System.err.println(e);
			}
		
		}//run
		
		private void sendFingerTable(String input)
		{
			String[] str = input.split(":");
			try{
				socket = new Socket(str[0],Integer.parseInt(str[1]));
				out = new ObjectOutputStream(new BufferedOutputStream(socket.getOutputStream()));
				out.writeObject(ft.getFinger());
			}
			catch(Exception e)
			{
				System.err.println(e);
			}
		}
		
		private void sendAddFT(String input)
		{
			if(addFT == true)
			{
				try{
					//gets the 1st successor
					String[] str = ft.get(1).split(":");
					socket = new Socket(str[0],Integer.parseInt(str[1]));
					out = new ObjectOutputStream(new BufferedOutputStream(socket.getOutputStream()));
					out.writeObject(input);
					seenAddFT++;
					if(seenAddFT == 2)
					{
						debug("addFT in if we have seen the add ft message twice");
						addFT = false;
						seenAddFT = 0;
					}
					
					socket.close();
					out.close();
						
				}
				catch(Exception e){
					System.err.println(e);
				}
			}
			
		}
		
		private void nextTwoMessagesAreObjects()
		{
			try{
				//Receives key then value
				key = in.readObject();
				value = in.readObject();
				in.close();
				
				}
			catch(Exception e)
			{
				System.err.println(e);
			}
		}
		private void nextMessageIsObject()
		{
			try{
				lookupReturn = in.readObject();
				in.close();
				
			}
			catch(Exception e)
			{
				System.err.println(e);
			}
		}
	}//myRunnable
}//class